package rx

/**
 * Created by CJJ on 2018/1/3.
 *
 */
fun main(args: Array<String>) {
    map()
}


fun map() {

    Observable.create<String>(object : Observable.OnSubscribe<String> {
        override fun call(t: Subscriber<in String>) {
            val end = 10
            for (i in 0..end)
                t.onNext(i.toString())
        }

    }).map {
        return@map it.takeLast(5)
    }.subscribe(object : Subscriber<String>() {
        override fun onNext(t: String) {
            println(t)
        }

        override fun onError(e: Throwable) {

        }

        override fun onCompleted() {

        }

    })
}